/*
 * University of British Columbia
 * Department of Computer Science
 * CPSC317 - Internet Programming
 * Assignment 2
 * 
 * Author: Jonatan Schroeder
 * January 2013
 * 
 * This code may not be used without written consent of the authors, except for 
 * current and future projects and assignments of the CPSC317 course at UBC.
 */

package ubc.cs317.rtsp.client.net;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

import ubc.cs317.rtsp.client.exception.RTSPException;
import ubc.cs317.rtsp.client.model.Frame;
import ubc.cs317.rtsp.client.model.Session;

/**
 * This class represents a connection with an RTSP server.
 */

/*
 * Problems to fix: - too small a gap between frames - too large a gap between
 * frames - frames which are out of order - lost frames
 */

public class RTSPConnection {

	public enum VideoState {
		PLAYING, STOPPED, PAUSED
	};

	private static final int BUFFER_LENGTH = 15000;
	private static final long MINIMUM_DELAY_READ_PACKETS_MS = 20;
	BlockingQueue<Frame> frameQueue;
	Thread frameQueueReaderThread;
	VideoState state;
	FrameQueueReader frameQueueReader;

	private Session session;
	private Timer rtpTimer;

	// TODO Add additional fields, if necessary
	private Socket socket;
	private PrintWriter socketWriter = null;
	private BufferedReader socketReader = null;
	private DatagramSocket datagramSocket;

	private int countSeq = 1;
	private int sessionID = 0;

	private String currentVideo;

	// fields for collecting statistics

	private static int firstSequenceNum = 0;
	private static int framesReceived = 0;
	private static int framesOutOfOrder = 0;
	private static int lastSequenceNum = -1;
	private static long startTime = 0;
	private static long lastTime = 0;
	private static long timeElapsed = 0;
	private static int framesLost = 0;

	/**
	 * Establishes a new connection with an RTSP server. No message is sent at
	 * this point, and no stream is set up.
	 * 
	 * @param session
	 *            The Session object to be used for connectivity with the UI.
	 * @param server
	 *            The hostname or IP address of the server.
	 * @param port
	 *            The TCP port number where the server is listening to.
	 * @throws RTSPException
	 *             If the connection couldn't be accepted, such as if the host
	 *             name or port number are invalid or there is no connectivity.
	 */
	public RTSPConnection(Session session, String server, int port)
			throws RTSPException {

		this.session = session;
		state = VideoState.STOPPED;

		try {
			socket = new Socket(server, port);
			socketWriter = new PrintOutWriter(socket.getOutputStream(), true);
			socketReader = new BufferedReader(new InputStreamReader(
					socket.getInputStream()));
		} catch (UnknownHostException e) {
			System.err.println("Don't know about host: " + server + ".");
			throw new RTSPException(server);
		} catch (IOException e) {
			System.err.println("Couldn't get I/O for the connection to: "
					+ server + ".");
			throw new RTSPException(server);
		}
	}

	/**
	 * Sends a SETUP request to the server. This method is responsible for
	 * sending the SETUP request, receiving the response and retrieving the
	 * session identification to be used in future messages. It is also
	 * responsible for establishing an RTP datagram socket to be used for data
	 * transmission by the server. The datagram socket should be created with a
	 * random UDP port number, and the port number used in that connection has
	 * to be sent to the RTSP server for setup. This datagram socket should also
	 * be defined to timeout after 1 second if no packet is received.
	 * 
	 * @param videoName
	 *            The name of the video to be setup.
	 * @throws RTSPException
	 *             If there was an error sending or receiving the RTSP data, or
	 *             if the RTP socket could not be created, or if the server did
	 *             not return a successful response.
	 */
	public synchronized void setup(String videoName) throws RTSPException {

		currentVideo = videoName;

		// setup datagramSocket
		try {
			datagramSocket = new DatagramSocket();
			datagramSocket.setSoTimeout(1000);
		} catch (SocketException e1) {
			e1.printStackTrace();
			throw new RTSPException("exception in setup");
		}

		// send SETUP request
		sendSetupRequest(currentVideo);

		// parse response
		RTSPResponse resp = readResponse();
		sessionID = resp.getSessionId();

		state = VideoState.STOPPED;

		// setup queue and reader
		frameQueue = new PriorityBlockingQueue<Frame>();
		frameQueueReader = new FrameQueueReader();
		frameQueueReaderThread = new Thread(frameQueueReader);
		frameQueueReaderThread.start();

	}

	private RTSPResponse readResponse() throws RTSPException {
		try {
			RTSPResponse resp = RTSPResponse.readRTSPResponse(socketReader);
			resp.print();
			if (resp.getResponseCode() != 200) {
				throw new RTSPException("response code not OK");
			}
			return resp;
		} catch (IOException e) {
			e.printStackTrace();
			throw new RTSPException("error parsing response");
		}

	}

	/**
	 * Sends a PLAY request to the server. This method is responsible for
	 * sending the request, receiving the response and, in case of a successful
	 * response, starting the RTP timer responsible for receiving RTP packets
	 * with frames.
	 * 
	 * @throws RTSPException
	 *             If there was an error sending or receiving the RTSP data, or
	 *             if the server did not return a successful response.
	 */
	public synchronized void play() throws RTSPException {

		sendPlayRequest();
		readResponse();
		// change state
		state = VideoState.PLAYING;
		// start threads for reading queue/reading packets
		startRTPTimer();

		synchronized (frameQueueReader) {
			frameQueueReader.notifyAll();
		}

	}

	/**
	 * Starts a timer that reads RTP packets repeatedly. The timer will wait at
	 * least MINIMUM_DELAY_READ_PACKETS_MS after receiving a packet to read the
	 * next one.
	 */
	private void startRTPTimer() {

		rtpTimer = new Timer();
		rtpTimer.schedule(new TimerTask() {
			@Override
			public void run() {
				receiveRTPPacket();
			}
		}, 0, MINIMUM_DELAY_READ_PACKETS_MS);
		startTime = System.currentTimeMillis();
	}

	/**
	 * Receives a single RTP packet and processes the corresponding frame. The
	 * data received from the datagram socket is assumed to be no larger than
	 * BUFFER_LENGTH bytes. This data is then parsed into a Frame object (using
	 * the parseRTPPacket method) and the method session.processReceivedFrame is
	 * called with the resulting packet. In case of timeout no exception should
	 * be thrown and no frame should be processed.
	 */
	private void receiveRTPPacket() {
		if (!datagramSocket.isClosed()) {

			byte[] buf = new byte[BUFFER_LENGTH];
			DatagramPacket packet = new DatagramPacket(buf, buf.length);
			try {
				receivePacket(packet);
			} catch (IOException e) {
				System.out.println("Error in receiving packet");
				e.printStackTrace();
				return;
			}
			byte[] received = packet.getData();

			Frame frame = parseRTPPacket(received);

			// collect statistics
			short sequenceNumber = frame.getSequenceNumber();

			long currentTime = System.currentTimeMillis();
			long packetInterval_ms = currentTime - lastTime;
			long time_difference_ms = MINIMUM_DELAY_READ_PACKETS_MS
					- packetInterval_ms;

			// pause if packet interval too small
			/*
			 * if (time_difference_ms > 0) { try {
			 * Thread.sleep(time_difference_ms); packetInterval_ms =
			 * System.currentTimeMillis() - lastTime; } catch
			 * (InterruptedException e) { System.out.println("poor show.");
			 * e.printStackTrace(); } }
			 */

			if (sequenceNumber != lastSequenceNum + 1) {
				framesOutOfOrder++;
			}
			lastSequenceNum = sequenceNumber;
			lastTime = currentTime;
			framesReceived++;
			// System.out.println("in: " + sequenceNumber + " at " +
			// packetInterval_ms);

			// session.processReceivedFrame(frame);
			frameQueue.add(frame);
		}
	}

	int timeOuts = 0;

	static final int STREAM_END_THRESHOLD = 5;

	private synchronized void receivePacket(DatagramPacket packet)
			throws IOException {
		try {
			datagramSocket.receive(packet);
			timeOuts = 0;
		} catch (SocketTimeoutException ex) {
			// Don't care about timeout
			if (state == VideoState.PLAYING) {
				timeOuts++;
				if (timeOuts == STREAM_END_THRESHOLD) {
					try {
						teardown();
					} catch (RTSPException e) {
						e.printStackTrace();
					}
				}
			}
		}
	}

	/**
	 * Sends a PAUSE request to the server. This method is responsible for
	 * sending the request, receiving the response and, in case of a successful
	 * response, cancelling the RTP timer responsible for receiving RTP packets
	 * with frames.
	 * 
	 * @throws RTSPException
	 *             If there was an error sending or receiving the RTSP data, or
	 *             if the server did not return a successful response.
	 */
	public synchronized void pause() throws RTSPException {

		sendPauseRequest();
		readResponse();
		// change state
		state = VideoState.PAUSED;
		cancelTimer();
	}

	private void cancelTimer() {
		rtpTimer.cancel();
		timeElapsed += (System.currentTimeMillis() - startTime);
	}

	/**
	 * Sends a TEARDOWN request to the server. This method is responsible for
	 * sending the request, receiving the response and, in case of a successful
	 * response, closing the RTP socket. This method does not close the RTSP
	 * connection, and a further SETUP in the same connection should be
	 * accepted. Also this method can be called both for a paused and for a
	 * playing stream, so the timer responsible for receiving RTP packets will
	 * also be cancelled.
	 * 
	 * @throws RTSPException
	 *             If there was an error sending or receiving the RTSP data, or
	 *             if the server did not return a successful response.
	 */
	public synchronized void teardown() throws RTSPException {

		cancelTimer();
		sendTeardownRequest();
		readResponse();

		closeDatagramSocket();
		// change state
		state = VideoState.STOPPED;
		frameQueueReaderThread.interrupt();
		frameQueue.clear();
		printStats();
		resetStats();
	}

	private void resetStats() {
		firstSequenceNum = 0;
		framesReceived = 0;
		framesOutOfOrder = 0;
		lastSequenceNum = -1;
		startTime = 0;
		timeElapsed = 0;
		framesLost = 0;
	}

	private void printStats() {
		int framesExpected = 1 + lastSequenceNum - firstSequenceNum;
		int framesLost = framesExpected - framesReceived;

		System.out.println("Frames received: " + framesReceived);
		System.out.println("Frames expected: " + framesExpected);
		System.out.println("Average frame rate: "
				+ ((double) framesReceived * 1000 / timeElapsed) + "f/s");
		System.out.println("Average out of order: "
				+ ((double) framesOutOfOrder / framesReceived));
		System.out.println("Average frame loss: "
				+ ((double) framesLost / framesReceived));
	}

	private synchronized void closeDatagramSocket() {
		datagramSocket.close();
	}

	/**
	 * Closes the connection with the RTSP server. This method should also close
	 * any open resource associated to this connection, such as the RTP
	 * connection, if it is still open.
	 */
	public synchronized void closeConnection() {
		if (datagramSocket == null) {
			return;
		}
		if (!datagramSocket.isClosed()) {
			datagramSocket.close();
		}
		while (!socket.isClosed()) {
			try {
				socket.close();
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("Could not close socket.");
			}
		}
	}

	/**
	 * Parses an RTP packet into a Frame object.
	 * 
	 * @param packet
	 *            the byte representation of a frame, corresponding to the RTP
	 *            packet.
	 * @return A Frame object.
	 */
	private static Frame parseRTPPacket(byte[] packet) {

		/*
		 * 0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0
		 * 1 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
		 * |V=2|P|X| CC |M| PT | sequence number |
		 * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
		 * timestamp |
		 * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
		 * synchronization source (SSRC) identifier |
		 * +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ |
		 * contributing source (CSRC) identifiers | | .... |
		 * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
		 * 
		 * byte: payloadType boolean: marker short: sequenceNumber int:
		 * timestamp byte[]: payload
		 */
		int length = packet.length;

		if (length < 12) {
			System.err.println("O NOES less than 12 byte packet");
			return null;
		}

		byte payloadType = (byte) (packet[1] & 0x0000007f);
		boolean isMarker = packet[1] >> 7 != 0;

		short sequenceNumber = (short) ((packet[3] & 0xFF) | (packet[2] << 8));
		assert (sequenceNumber >= 0);

		// Integers holding the bits in the timestamp packet bytes to prevent
		// signing
		int[] inTimeStamp = new int[4];

		for (int x = 0; x < 4; x++) {
			if (packet[4 + x] < 0) {
				inTimeStamp[x] = packet[4 + x] + 256;
			} else {
				inTimeStamp[x] = packet[4 + x];
			}
		}

		// Timestamp is big-endian
		int timestamp = inTimeStamp[0] << (8 * 3) | inTimeStamp[1] << (8 * 2)
				| inTimeStamp[2] << 8 | inTimeStamp[3];

		int csrcCount = packet[0] & 0x0000000f;
		int headerLength = 12 + csrcCount;
		assert (packet.length - headerLength > 0);
		byte[] payload = new byte[packet.length - headerLength];
		int i = 0;
		while (i < packet.length - headerLength) {
			payload[i] = packet[i + headerLength];
			i++;
		}

		return new Frame(payloadType, isMarker, sequenceNumber, timestamp,
				payload);
	}

	// helper functions added
	private void sendSetupRequest(String currentVideo) {
		socketWriter.println("SETUP " + currentVideo + " RTSP/1.0");
		socketWriter.println("CSeq: " + countSeq++);
		socketWriter.println("Transport: RTP/UDP; client_port= "
				+ datagramSocket.getLocalPort());
		socketWriter.println();
		socketWriter.flush();
	}

	private void sendPlayRequest() {
		socketWriter.println("PLAY " + currentVideo + " RTSP/1.0");
		socketWriter.println("CSeq: " + countSeq++);
		socketWriter.println("Session: " + sessionID);
		socketWriter.println();
		socketWriter.flush();
	}

	private void sendPauseRequest() {
		socketWriter.println("PAUSE " + currentVideo + " RTSP/1.0");
		socketWriter.println("CSeq: " + countSeq++);
		socketWriter.println("Session: " + sessionID);
		socketWriter.println();
		socketWriter.flush();
	}

	private void sendTeardownRequest() {
		socketWriter.println("TEARDOWN " + currentVideo + " RTSP/1.0");
		socketWriter.println("CSeq: " + countSeq++);
		socketWriter.println("Session: " + sessionID);
		socketWriter.println();
		socketWriter.flush();
	}

	/**
	 * Reads from frame queue and processes frame at a constant rate. If
	 * out-of-order frame is encountered, thread repeatedly pauses and resumes.
	 * Finally it considers this frame lost and changes its expectations.
	 * 
	 */
	private class FrameQueueReader implements Runnable {

		private static final long PACKET_TIMEOUT = MINIMUM_DELAY_READ_PACKETS_MS;
		private static final long MAX_PACKET_TIMEOUT = PACKET_TIMEOUT * 5;
		private static final long BUFFER_EMPTY_TIMEOUT_MS = 2000;
		private static final int MAX_TIMEOUTS = 10;
		private long lastTimestamp;
		private int lastSeqNumber;
		private int timeouts;

		public FrameQueueReader() {
			lastSeqNumber = 0;
			lastTimestamp = 0;
			timeouts = 0;
		}

		@Override
		public void run() {
			while (true) {
				while (state != VideoState.PLAYING) {

					try {
						synchronized (this) {
							System.out.println("waiting!");
							wait();
						}
					} catch (InterruptedException e) {
						System.err.println("queue reader interrupted");
						return;
					}

					// System.out.println("frameReader: video in wrong state. returning");
					// return;
				}
				// if queue is empty: fill queue with packets
				if (frameQueue.size() == 0) {
					if (timeouts < MAX_TIMEOUTS) {
						try {
							System.out.println("buffer is empty");
							Thread.sleep(BUFFER_EMPTY_TIMEOUT_MS);
						} catch (InterruptedException e) {
							System.err.println("queue reader interrupted");
							return;
						}
						timeouts++;
					} else {
						try {
							teardown();
						} catch (RTSPException e) {
							System.err
									.println("queue reader couldn't send teardown request");
							return;
						}
					}
				} else {
					Frame frame;
					try {
						frame = frameQueue.take();
						lastSeqNumber = frame.getSequenceNumber();
						long newTimestamp = frame.getTimestamp();
						long interval = newTimestamp - lastTimestamp;
						if (interval > 0) {
							Thread.sleep(interval);
							session.processReceivedFrame(frame);
							System.out.println("out: " + lastSeqNumber + " at "
									+ interval);
							lastTimestamp = newTimestamp;
						}
					} catch (InterruptedException e) {
						System.err.println("queue reader interrupted");
						return;
					}
				}
			}
		}
	}
}